package com.atguigu.flink.watermark;

import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * Created by Smexy on 2022/11/23
 *
 *   模拟乱序的情况。
 *          一旦窗口关闭，那么迟到的数据，无法运算的。
 *
 *          如何解决迟到的数据？
 *              三板斧。
 *                  第一板斧:    推迟水印(时钟)的推进，类似把表调慢(添加一个乱序容忍度)！
 *                  第二板斧：   延迟窗口的关闭时间。
 *                                  到点后，窗口立刻触发运算。
 *                                  但是运算后，不关闭销毁，给你一定时间，在这个时间内，只要能进入窗口，会立刻为你再算一次。
 *                  第三板斧:    窗口关闭后的数据，使用一个侧流接收。随后处理侧流！
 *
 *
 *             还不行？
 *                  放弃吧！ 你的场景，不适合流式计算！建议使用批处理！
 *
 *
 *
 */
public class Demo3_Orderness
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.getConfig().setAutoWatermarkInterval(2000);

        OutputTag<WaterSensor> outputTag = new OutputTag<>("lateData", TypeInformation.of(WaterSensor.class));

        env.setParallelism(1);

        //水印策略
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            //会把水印延迟3秒推进
            .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
             .withTimestampAssigner((e, r) -> e.getTs());

        SingleOutputStreamOperator<String> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy)
            //第一个窗口:  [0,4999] 或 [0,5000)
            //第二个窗口:  [5000,9999]
            .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
            //延迟窗口的关闭
            .allowedLateness(Time.seconds(3))
            //使用侧流接收窗口关闭后的迟到数据
            .sideOutputLateData(outputTag)
            .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>()
            {
                @Override
                public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                    TimeWindow window = context.window();

                    out.collect(window + "数据:" + MyUtil.parseList(elements).toString());
                }
            });


        //操作主流
            ds.print();
        //操作侧流
            ds.getSideOutput(outputTag).printToErr();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
